package com.fang.dubbo.mock.rpc.netty;


import com.fang.dubbo.mock.common.URL;
import com.fang.dubbo.mock.rpc.protocol.Invocation;
import com.fang.dubbo.mock.rpc.register.LoadBalance;
import com.fang.dubbo.mock.rpc.register.RemoteMapRegister;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Proxy;
import java.util.List;
import java.util.concurrent.*;

/**
 * @author fanglingxiao
 * @version 1.0
 * @description TODO
 * @date 2022/1/2 1:28 下午
 **/
@Slf4j
public class NettyClient {
    private static final ThreadFactory NAME_FACTORY = new ThreadFactoryBuilder().setNameFormat("fang-test-%d").build();

    private static final ExecutorService EXECUTOR = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), 200,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(1024), NAME_FACTORY, new ThreadPoolExecutor.AbortPolicy());
    private static NettyClientHandler clientHandler;

    /**
     * 编写方法使用代理对象，获取代理对象
     */
    public <T> T getProxy(Class<?> clazz) {
        // !!!
        RemoteMapRegister.register();

        return (T) Proxy.newProxyInstance(
                Thread.currentThread().getContextClassLoader(),
                new Class[]{clazz},
                (proxy, method, args) -> {
                    // 去注册中心获得服务地址列表
                    List<URL> urlList = RemoteMapRegister.getRegister(clazz.getName());
                    //负载均衡
                    URL url = LoadBalance.random(urlList);
                    if (clientHandler == null) {
                        initClient(url);
                    }
                    log.info("NettyClient method={},args={}", method, args);
                    Invocation invocation = new Invocation(clazz.getName(), method.getName(), method.getParameterTypes(), args);
                    // 设置参数协议参数
                    clientHandler.setParam(invocation);
                    return EXECUTOR.submit(clientHandler).get();
                });
    }

    /**
     * 初始化客户端
     */
    private static void initClient(URL url) {
        clientHandler = new NettyClientHandler();
        // 创建EventLoopGroup
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel sc) throws Exception {
                        ChannelPipeline pipeline = sc.pipeline();
                        pipeline.addLast(new ObjectDecoder(1024 * 1024,
                                ClassResolvers.cacheDisabled(this.getClass().getClassLoader())));
                        pipeline.addLast(new ObjectEncoder());
                        pipeline.addLast(clientHandler);
                    }
                });
        try {
            log.info("invoke {}", url);
            bootstrap.connect(url.getUsername(), url.getPort()).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
